package com.wiloon.java8x.thread.pool;

import org.apache.commons.lang3.RandomUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by wiloon on 6/15/17;
 */
public class PoolX {
    static Logger logger = LoggerFactory.getLogger(PoolX.class);
    static int inc = 0;
    static int dec = 0;
    static ThreadPoolExecutor threadPoolExecutor;
    static AtomicInteger taskCount = new AtomicInteger(0);
    static int monitorDuration = 3;

    public static void main(String[] args) {

        new Timer().schedule(new Monitor(), 0, TimeUnit.SECONDS.toMillis(monitorDuration));
        new Thread(new Worker()).start();
        new Thread(new PressController()).start();
    }

    static class PressController implements Runnable {

        @Override
        public void run() {
            for (; ; ) {
                try {
                    int in = System.in.read();
                    if (in == 10) continue;
                    logger.info("" + in);
                    if (in == 119) inc = 1;
                    if (in == 115) dec = 1;
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    static class Worker implements Runnable {
        @Override
        public void run() {
            threadPoolExecutor = new ThreadPoolExecutor(
                    1,
                    3,
                    2,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(10));

            int count = 1;
            for (; ; ) {
                for (int i = 0; i < count; i++) {
                    try {
                        threadPoolExecutor.execute(new Task());
                    } catch (RejectedExecutionException e) {
                        logger.error(e.getMessage());
                    }
                }

                try {
                    TimeUnit.SECONDS.sleep(1);
                    logger.info("producer:{}", count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (inc == 1) {
                    count += 1;
                    inc = 0;
                }
                if (dec == 1) {
                    count -= 1;
                    dec = 0;
                }
            }
        }
    }


    static class Task implements Runnable {

        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(RandomUtils.nextInt(0, 1000));
                taskCount.incrementAndGet();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Monitor extends TimerTask {
        int lastTaskCount = 0;

        @Override
        public void run() {
            if (threadPoolExecutor == null) return;
            logger.info("queue:{}, pool size:{}, active thread:{}, largest pool size:{},consumer rate:{}, task:{}, completed:{}",
                    threadPoolExecutor.getQueue().size(),
                    threadPoolExecutor.getPoolSize(),
                    threadPoolExecutor.getActiveCount(),
                    threadPoolExecutor.getLargestPoolSize(),
                    (taskCount.get() - lastTaskCount) / monitorDuration,
                    threadPoolExecutor.getTaskCount(),
                    threadPoolExecutor.getCompletedTaskCount()
            );

            lastTaskCount = taskCount.get();
        }
    }
}
